package cn.doitedu.etl;

import cn.doitedu.udfs.GpsToGeohashFunction;
import cn.doitedu.udfs.MapToJsonFunction;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job01_UserActionLog_ODS_DWD {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        //env.setRuntimeMode(RuntimeExecutionMode.BATCH);  // 以“批计算模式”来运行本job
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射ods层的kafka中的用户行为日志topic
        tenv.executeSql(
                " CREATE TABLE ods_user_action_kafka(                         "+
                        "   username STRING,                                   "+
                        " 	release_channel STRING,                            "+
                        " 	app_version STRING,                                "+
                        " 	app_id STRING,                                     "+
                        " 	device_type STRING,                                "+
                        " 	os_name STRING,                                    "+
                        " 	os_version STRING,                                 "+
                        " 	screen_resolution STRING,                          "+
                        " 	session_id STRING,                                 "+
                        " 	lat DOUBLE,                                        "+
                        " 	lng DOUBLE,                                        "+
                        " 	eventId STRING,                                    "+
                        " 	eventTime BIGINT,                                  "+
                        " 	properties MAP<STRING,STRING>,                     "+
                        " 	pt AS proctime()                                   "+
                        " ) with (                                             "+
                        "   'connector' = 'kafka',                             "+
                        "   'topic' = 'ods-user-action',                       "+
                        "   'properties.bootstrap.servers' = 'doitedu:9092',   "+
                        "   'properties.group.id' = 'g008',                    "+
                        "   'scan.startup.mode' = 'latest-offset',             "+
                        "   'value.format' = 'json',                           "+
                        "   'value.fields-include' = 'EXCEPT_KEY'              "+
                        " )                                                    "
        );

        // 建表，映射hbase中的用户注册信息维表
        tenv.executeSql(
                " CREATE TABLE ums_member_hbase (        "+
                        "  username STRING,                      "+
                        "  f ROW<                                "+
                        "   id bigint,                           "+
                        " 	member_level_id bigint,              "+
                        " 	phone string,                        "+
                        " 	status int,                          "+
                        "   create_time timestamp(3),            "+
                        " 	gender int,                          "+
                        " 	birthday date,                       "+
                        " 	city string,                         "+
                        " 	job string,                          "+
                        " 	source_type int,                     "+
                        " 	integration int,                     "+
                        " 	growth int,                          "+
                        " 	luckey_count int,                    "+
                        " 	history_integration int,             "+
                        " 	modify_time timestamp(3)>,           "+
                        "   primary key(username) not enforced	 "+
                        " ) WITH (                               "+
                        "  'connector' = 'hbase-2.2',            "+
                        "  'table-name' = 'dim_ums_member',      "+
                        "  'zookeeper.quorum' = 'doitedu:2181'   "+
                        " )                                      "
        );


        // 建表，映射hbase中的地理位置信息维表
        tenv.executeSql(
                " CREATE TABLE dim_reference_geo_hbase (        "+
                        "  geohash STRING,               "+
                        "  f ROW<                        "+
                        "       province string,         "+
                        " 	    city string,             "+
                        " 	    region string            "+
                        " 	>,                           "+
                        "   primary key(geohash) not enforced	 "+
                        " ) WITH (                               "+
                        "  'connector' = 'hbase-2.2',            "+
                        "  'lookup.async' = 'true',              "+  // 开启lookup异步查询
                        "  'lookup.cache' = 'PARTIAL',              "+  // 开启lookup缓存
                        "  'lookup.partial-cache.max-rows' = '100000',   "+  // 开启lookup缓存
                        "  'lookup.partial-cache.expire-after-access' = '10 m',   "+  // 开启lookup缓存的生命周期，且在访问后会计时重置
                        "  'lookup.partial-cache.caching-missing-key' = 'true',   "+  // 缓存缺失维度值的key
                        "  'table-name' = 'dim_reference_geo',   "+
                        "  'zookeeper.quorum' = 'doitedu:2181'   "+
                        " )                                      "
        );


        /**
         * put 'dim_url_info','/mall/','f:page_type','商品详情页'
         * put 'dim_url_info','/mall/','f:page_business','商城服务'
         * put 'dim_url_info','/mall/','f:page_channel','A'
         * put 'dim_url_info','/mall/','f:page_lanmu','C'
         */
        // 建表，映射hbase中的页面信息维表
        tenv.executeSql(
                " CREATE TABLE dim_url_info_hbase (        "+
                        "  url_prefix STRING,            "+
                        "  f ROW<                        "+
                        "       page_type string,        "+
                        " 	    page_business string,    "+
                        " 	    page_channel string,     "+
                        " 	    page_lanmu string        "+
                        " 	>,                           "+
                        "   primary key(url_prefix) not enforced "+
                        " ) WITH (                               "+
                        "  'connector' = 'hbase-2.2',            "+
                        "  'lookup.async' = 'true',              "+  // 开启lookup异步查询
                        "  'lookup.cache' = 'PARTIAL',              "+  // 开启lookup缓存
                        "  'lookup.partial-cache.max-rows' = '100000',   "+  // 开启lookup缓存
                        "  'lookup.partial-cache.expire-after-access' = '10 m',   "+  // 开启lookup缓存的生命周期，且在访问后会计时重置
                        "  'lookup.partial-cache.caching-missing-key' = 'true',   "+  // 缓存缺失维度值的key
                        "  'table-name' = 'dim_url_info',        "+
                        "  'zookeeper.quorum' = 'doitedu:2181'   "+
                        " )                                      "
        );



        // 建表，映射dwd层的kafka中的宽表topic
        tenv.executeSql(
                " CREATE TABLE dwd_user_action_kafka (                            "+
                        "   user_name STRING,                                              "+
                        " 	release_channel STRING,                                       "+
                        " 	app_version STRING,                                           "+
                        " 	app_id STRING,                                                "+
                        " 	device_type STRING,                                           "+
                        " 	os_name STRING,                                               "+
                        " 	os_version STRING,                                            "+
                        " 	screen_resolution STRING,                                     "+
                        " 	session_id STRING,                                            "+
                        " 	lat DOUBLE,                                                   "+
                        " 	lng DOUBLE,                                                   "+
                        " 	event_id STRING,                                               "+
                        " 	action_time BIGINT,                                             "+
                        " 	properties MAP<STRING,STRING>,                                "+
                        "   user_id bigint,                                                    "+
                        " 	member_level_id bigint,                                       "+
                        " 	phone string,                                                 "+
                        " 	status int,                                                   "+
                        "   create_time timestamp(3),                                     "+
                        " 	gender int,                                                   "+
                        " 	birthday date,                                                "+
                        " 	register_city string,                                         "+
                        " 	job string,                                                   "+
                        " 	source_type int,                                              "+
                        " 	integration int,                                              "+
                        " 	growth int,                                                   "+
                        " 	luckey_count int,                                             "+
                        " 	history_integration int,                                      "+
                        " 	modify_time timestamp(3),                                     "+
                        "   province string,                                              "+
                        " 	city string,                                                  "+
                        " 	region string,                       				          "+
                        "   page_type string,                                             "+
                        " 	page_business string,                                         "+
                        " 	page_channel string,                                          "+
                        " 	page_lanmu string                                             "+
                        " ) WITH (                                                        "+
                        "     'connector' = 'kafka',                                      "+
                        "     'topic' = 'dwd-user-action',                                "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',            "+
                        "     'properties.group.id' = 'g008',                             "+
                        "     'scan.startup.mode' = 'latest-offset',                      "+
                        "     'value.format' = 'json',                                    "+
                        "     'value.fields-include' = 'EXCEPT_KEY'                       "+
                        " )				                                                  "
        );



        // 写一个sql,进行各种维表的关联
        tenv.createTemporaryFunction("gps2geo", GpsToGeohashFunction.class);
        tenv.executeSql(
                        " CREATE TEMPORARY VIEW joined_result AS                                                                                                 "+
                        // 因为一个新用户刚注册时产生的注册信息，到达hbase的时间，可能略晚于这个人的第一条行为数据到达时间，所以需要开启延迟重试
                        " SELECT  /*+ LOOKUP('table'='ums_member_hbase', 'retry-predicate'='lookup_miss', 'retry-strategy'='fixed_delay', 'fixed-delay'='100ms','max-attempts'='3') */        "+
                        "  a.username    as user_name                                                                                                            "+
                        " ,a.release_channel                                                                                                                     "+
                        " ,a.app_version                                                                                                                         "+
                        " ,a.app_id                                                                                                                              "+
                        " ,a.device_type                                                                                                                         "+
                        " ,a.os_name                                                                                                                             "+
                        " ,a.os_version                                                                                                                          "+
                        " ,a.screen_resolution                                                                                                                   "+
                        " ,a.session_id                                                                                                                          "+
                        " ,a.lat                                                                                                                                 "+
                        " ,a.lng                                                                                                                                 "+
                        " ,a.eventId     as event_id                                                                                                             "+
                        " ,a.eventTime   as action_time                                                                                                          "+
                        " ,a.properties                                                                                                                          "+
                        " ,b.id   as user_id                                                                                                                     "+
                        " ,b.member_level_id                                                                                                                     "+
                        " ,b.phone                                                                                                                               "+
                        " ,b.status                                                                                                                              "+
                        " ,b.create_time                                                                                                                         "+
                        " ,b.gender                                                                                                                              "+
                        " ,b.birthday                                                                                                                            "+
                        " ,b.city as register_city                                                                                                               "+
                        " ,b.job                                                                                                                                 "+
                        " ,b.source_type                                                                                                                         "+
                        " ,b.integration                                                                                                                         "+
                        " ,b.growth                                                                                                                              "+
                        " ,b.luckey_count                                                                                                                        "+
                        " ,b.history_integration                                                                                                                 "+
                        " ,b.modify_time                                                                                                                         "+
                        " ,c.province                                                                                                                            "+
                        " ,c.city                                                                                                                                "+
                        " ,c.region                                                                                                                              "+
                        " ,d.page_type                                                                                                                           "+
                        " ,d.page_business                                                                                                                       "+
                        " ,d.page_channel                                                                                                                        "+
                        " ,d.page_lanmu                                                                                                                          "+
                        " FROM ods_user_action_kafka a                                                                                                           "+
                        " LEFT JOIN ums_member_hbase         FOR SYSTEM_TIME AS OF a.pt  as b ON  a.username = b.username                                        "+
                        " LEFT JOIN dim_reference_geo_hbase  FOR SYSTEM_TIME AS OF a.pt  as c ON gps2geo(a.lat,a.lng)=c.geohash                                  "+
                        " LEFT JOIN dim_url_info_hbase       FOR SYSTEM_TIME AS OF a.pt  as d ON regexp_extract(a.properties['url'],'(^/.*/).*?')=d.url_prefix   "
        );

        // 把关联结果写入dwd层的kafka topic
        tenv.executeSql("insert into dwd_user_action_kafka select * from joined_result");


        // 建表，映射 doris中的用户行为明细宽表
        tenv.executeSql(
                " create table user_action_detail_doris (                  "+
                        "     event_id                  string                     "+
                        "     ,user_name                string                     "+
                        "     ,release_channel          string                     "+
                        "     ,app_version              string                     "+
                        "     ,app_id                   string                     "+
                        "     ,device_type              string                     "+
                        "     ,os_name                  string                     "+
                        "     ,os_version               string                     "+
                        "     ,screen_resolution        string                     "+
                        "     ,session_id               string                     "+
                        "     ,lat                      double                     "+
                        "     ,lng                      double                     "+
                        "     ,action_time              bigint                     "+
                        "     ,properties_json          string                     "+
                        "     ,user_id                  bigint                     "+
                        "     ,member_level_id          bigint                     "+
                        "     ,phone                    string                     "+
                        "     ,status                   int                        "+
                        "     ,create_time              timestamp(3)               "+
                        "     ,gender                   int                        "+
                        "     ,birthday                 date                       "+
                        "     ,register_city            string                     "+
                        "     ,job                      string                     "+
                        "     ,source_type              int                        "+
                        "     ,integration              int                        "+
                        "     ,growth                   int                        "+
                        "     ,luckey_count             int                        "+
                        "     ,history_integration      int                        "+
                        "     ,modify_time              timestamp(3)               "+
                        "     ,province                 string                     "+
                        "     ,city                     string                     "+
                        "     ,region                   string                     "+
                        "     ,page_type                string                     "+
                        "     ,page_business            string                     "+
                        "     ,page_channel             string                     "+
                        "     ,page_lanmu               string                     "+
                        " 	,dt                       date                         "+
                        " ) with (                                                 "+
                        "        'connector' = 'doris',                            "+
                        "        'fenodes' = 'doitedu:8030',                       "+
                        "        'table.identifier' = 'dwd.user_action_detail',    "+
                        "        'username' = 'root',                              "+
                        "        'password' = 'root',                              "+
                        "        'sink.label-prefix' = 'doris_label-001'           "+
                        " )                                                        "
        );


        // 把关联结果写入dwd层的doris宽表
        tenv.createTemporaryFunction("map_to_json", MapToJsonFunction.class);
        tenv.executeSql(
                " INSERT INTO user_action_detail_doris                                            "+
                        " SELECT                                                                          "+
                        "  event_id                                                                       "+
                        " ,user_name                                                                      "+
                        " ,release_channel                                                                "+
                        " ,app_version                                                                    "+
                        " ,app_id                                                                         "+
                        " ,device_type                                                                    "+
                        " ,os_name                                                                        "+
                        " ,os_version                                                                     "+
                        " ,screen_resolution                                                              "+
                        " ,session_id                                                                     "+
                        " ,lat                                                                            "+
                        " ,lng                                                                            "+
                        " ,action_time                                                                    "+
                        " ,map_to_json(properties) as properties_json                                     "+
                        " ,user_id                                                                        "+
                        " ,member_level_id                                                                "+
                        " ,phone                                                                          "+
                        " ,status                                                                         "+
                        " ,create_time                                                                    "+
                        " ,gender                                                                         "+
                        " ,birthday                                                                       "+
                        " ,register_city                                                                  "+
                        " ,job                                                                            "+
                        " ,source_type                                                                    "+
                        " ,integration                                                                    "+
                        " ,growth                                                                         "+
                        " ,luckey_count                                                                   "+
                        " ,history_integration                                                            "+
                        " ,modify_time                                                                    "+
                        " ,province                                                                       "+
                        " ,city                                                                           "+
                        " ,region                                                                         "+
                        " ,page_type                                                                      "+
                        " ,page_business                                                                  "+
                        " ,page_channel                                                                   "+
                        " ,page_lanmu                                                                     "+
                        " ,to_date(date_format(to_timestamp_ltz(action_time,3),'yyyy-MM-dd'))  as dt      "+
                        " FROM joined_result                                                              "
        );




    }


}
